package com.aiwiown.snackmq.storage.service;

import com.aiwiown.snackmq.common.protocol.TopicPartition;
import com.aiwiown.snackmq.storage.partition.LogPartition;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * 零拷贝分区存储服务实现
 * 直接返回ByteBuffer切片，避免内存拷贝
 */
@Slf4j
public class ZeroCopyPartitionedStorageService implements ZeroCopyStorageService {

    private final Map<TopicPartition, LogPartition> sharedPartitions;

    public ZeroCopyPartitionedStorageService(Map<TopicPartition, LogPartition> sharedPartitions) {
        this.sharedPartitions = sharedPartitions;
    }

    public void start() {
        log.info("Zero-copy partitioned storage service started and is using shared partitions.");
    }

    public void close() throws IOException {
        log.info("Zero-copy partitioned storage service closed.");
    }

    @Override
    public List<ByteBuffer> readZeroCopy(TopicPartition partition, long offset, int maxMessages) throws IOException {
        if (partition == null) {
            throw new IllegalArgumentException("Partition cannot be null");
        }
        if (offset < 0) {
            throw new IllegalArgumentException("Offset cannot be negative");
        }
        if (maxMessages <= 0) {
            return new ArrayList<>();
        }

        LogPartition logPartition = sharedPartitions.get(partition);
        if (logPartition == null) {
            log.warn("Partition {} is not active on this broker", partition);
            return new ArrayList<>();
        }

        // 【代码修复】: 移除此处的冗余循环，直接委托给 LogPartition 的批量读取方法。
        // 这遵循了单一职责原则，让 LogPartition 负责其内部的读取逻辑。
        List<ByteBuffer> buffers = logPartition.readBatchZeroCopy(offset, maxMessages);
        log.debug("Read {} messages from partition {} starting at offset {}",
                buffers.size(), partition, offset);
        return buffers;
    }

    @Override
    public ByteBuffer readMessageZeroCopy(TopicPartition partition, long offset) throws IOException {
        if (partition == null) {
            throw new IllegalArgumentException("Partition cannot be null");
        }
        if (offset < 0) {
            throw new IllegalArgumentException("Offset cannot be negative");
        }

        LogPartition logPartition = sharedPartitions.get(partition);
        if (logPartition == null) {
            log.warn("Partition {} is not active on this broker", partition);
            return null;
        }
        return logPartition.readZeroCopy(offset);
    }

    @Override
    public List<ByteBuffer> readRangeZeroCopy(TopicPartition partition, long startOffset, long endOffset) throws IOException {
        if (partition == null) {
            throw new IllegalArgumentException("Partition cannot be null");
        }
        if (startOffset < 0 || endOffset < 0 || startOffset > endOffset) {
            throw new IllegalArgumentException("Invalid offset range [" + startOffset + ", " + endOffset + ")");
        }
        LogPartition logPartition = sharedPartitions.get(partition);
        if (logPartition == null) {
            log.warn("Partition {} is not active on this broker", partition);
            return new ArrayList<>();
        }

        // 【代码修复】: 同样，直接委托给 LogPartition，并计算 maxMessages。
        int maxMessages = (int) (endOffset - startOffset);
        List<ByteBuffer> buffers = logPartition.readBatchZeroCopy(startOffset, maxMessages);
        log.debug("Read {} messages from partition {} in range [{}, {})",
                buffers.size(), partition, startOffset, endOffset);
        return buffers;
    }
}